import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;

public class RASelection implements RAExpression {

	private final Map<String, TableData> catalog;
	
	// If evaluateSelectList is true then this RASelection will evaluate the select list
	// otherwise it will run columns to project.
	private boolean evaluateSelectList;
	private ArrayList<Expression> select;
	private ArrayList<Expression> columnsToProject;
	
	private final Map<String, String> from;
	private Expression where;
	private final String correlation;
	private final String table;
	
	public RASelection(final Map<String, TableData> catalog, final ArrayList<Expression> selectList, final Map<String, String> from,
			final String correlation, final String table, final Expression where, boolean evaluateSelectList)
	{
		this.evaluateSelectList = evaluateSelectList;
		
		this.catalog = catalog;
		this.select = selectList;
		this.from = from;
		this.where = where;
		
		this.correlation = correlation;
		this.table = table;
	}
	
	public RASelection(final Map<String, TableData> catalog, final ArrayList<Expression> selectList, final Map<String, String> from,
			final Expression where, boolean evaluateSelectList)
	{
		this.evaluateSelectList = evaluateSelectList;
		
		this.catalog = catalog;
		this.select = selectList;
		this.from = from;
		this.where = where;
		
		@SuppressWarnings("unchecked")
		Map.Entry<String, String> thisEntry = (Map.Entry<String, String>) from.entrySet().toArray()[0];
		this.correlation = thisEntry.getKey();
		this.table = thisEntry.getValue();
	}
	
	public RASelection(final SQLParser parser, final Map<String, TableData> catalog, boolean evaluateSelectList)
	{
		this(catalog, parser.getSELECT(), parser.getFROM(), parser.getWHERE(), evaluateSelectList);
	}

	@Override
	public RAOutput execute() throws IOException
	{
		if(DBConfig.DEBUG)
		{
			System.out.println("Executing RASelection: " + this.toString());
		}

		// ArrayList <Attribute> inAtts = new ArrayList <Attribute>();
		// inAtts.add (new Attribute ("Int", "o_orderkey"));
		// inAtts.add (new Attribute ("Int", "o_custkey"));
		// inAtts.add (new Attribute ("Str", "o_orderstatus"));
		// inAtts.add (new Attribute ("Float", "o_totalprice"));
		// inAtts.add (new Attribute ("Str", "o_orderdate"));
		// inAtts.add (new Attribute ("Str", "o_orderpriority"));
		// inAtts.add (new Attribute ("Str", "o_clerk"));
		// inAtts.add (new Attribute ("Int", "o_shippriority"));
		// inAtts.add (new Attribute ("Str", "o_comment"));

		// Input to a selection always comes directly from the base table.
		final TableData td = catalog.get(table);
		ArrayList<Attribute> inAtts = td.getTableLayout(correlation);

		// ArrayList <Attribute> outAtts = new ArrayList <Attribute> ();
		// outAtts.add (new Attribute ("Int", "att1"));
		// outAtts.add (new Attribute ("Float", "att2"));
		// outAtts.add (new Attribute ("Str", "att3"));
		// outAtts.add (new Attribute ("Int", "att4"));

		// HashMap <String, String> exprs = new HashMap <String, String> ();
		// exprs.put ("att1", "o_orderkey");
		// exprs.put ("att2", "(o_totalprice * Float (1.5)) + Int (1)");
		// exprs.put ("att3", "o_orderdate + Str (\" this is my string\")");
		// exprs.put ("att4", "o_custkey");
		
		// Initialize the list of output attribute names, data types, and evaluation expressions.
		ArrayList<Expression> selectList = (this.evaluateSelectList) ? select : columnsToProject;
		if(selectList == null)
		{
			selectList = catalog.get(table).getSelectList(correlation);
		}
		int selectSize = selectList.size();
		ArrayList<Attribute> outAtts = new ArrayList<Attribute>(selectSize);
		final HashMap<String, String> exprs = new HashMap<String, String>();
		ArrayList<Attribute> nextRAInputAttributeList = new ArrayList<Attribute>(selectSize);
		for(int i = 0; i < selectSize; i++)
		{
			String attName = "att" + i;
			Expression e = selectList.get(i);
			String returnType = e.getReturnType(from, catalog); 
			String executionString = e.toExecutionString();

			outAtts.add(new Attribute(returnType, attName));
			exprs.put(attName, executionString);
			nextRAInputAttributeList.add(new Attribute(returnType, executionString));
		}

		// String selection =
		// "o_orderdate > Str (\"1996-12-19\") && o_custkey < Int (100)";
		
		// Initialize the selection predicate.
		String selectionPredicate;
		if(where != null)
		{
			selectionPredicate = where.toExecutionString();
		}
		else
		{
			selectionPredicate = "true";
		}

		if(DBConfig.DEBUG)
		{
			System.out.println("selection predicate = " + selectionPredicate);
		}

		// run the selection operation
		
//		try
//		{
//			Selection foo = new Selection(inAtts, outAtts, selection, exprs, "orders.tbl", "out.tbl", "g++", "cppDir/");
//		}
//		catch(Exception e)
//		{
//			throw new RuntimeException(e);
//		}

		final String selectionInFile = table + ".tbl";
		final String selectionOutFile = table + ".sout";
		try
		{
			new Selection(inAtts, outAtts, selectionPredicate, exprs, selectionInFile,
					selectionOutFile, DBConfig.CPP_COMPILER, DBConfig.CPP_TEMP_DIR);
		}
		catch (final Exception e)
		{
			throw new RuntimeException(e);
		}
		
		return new RAOutput(selectionOutFile, nextRAInputAttributeList);
	}

	@Override
	public <T> T accept(final RAExpressionVisitor<T> v, final Object... args)
	{
		return v.selectionCase(this, args);
	}

	@Override
	public RAStatistics estimateCost(Map<String, TableData> catalog, Map<String, TableData> tempCatalog)
	{
		int card = catalog.get(table).getTupleCount();
		
		// Assuming all attributes are independent, then per selection predicate we multiple by a reduction factor.
		if(this.where != null)
		{
			Map<String, TableData> updatedTempCatalog = new HashMap<String, TableData>();
			double filterFactor = where.computeFilterFactor(from, catalog, tempCatalog, updatedTempCatalog);
			DBUtils.mergeCatalogs(tempCatalog, updatedTempCatalog);
			long cost = Math.round(card * filterFactor);
			
			DBUtils.setCatalogMaxCard(tempCatalog, cost);
			
			return new RAStatistics(cost, cost);
		}
		
		return new RAStatistics(card, card);
	}

	@Override
	public void getTables(Map<String, String> tables)
	{
		tables.put(correlation, table);
//		tables.putAll(from);
	}

	@Override
	public ArrayList<RAExpression> getJoinTransformations()
	{
		return null;
	}

	@Override
	public boolean referencedByExpression(Expression e)
	{
		return e.referencesTable(this.correlation);
	}

	@Override
	public void addWhereDisjunction(Expression e)
	{
		if(where == null)
		{
			where = e;
		}
		else
		{
			Expression newWhere = new Expression("and");
			newWhere.setSubexpression(where, e);
		}
	}

	@Override
	public String joinPlanToString()
	{
		return "(" + correlation + ")";
	}
	
	public String toString()
	{
		return joinPlanToString();
	}

	@Override
	public void pushdownReferencedColumnsList(ArrayList<Expression> referencedColumns)
	{
		if(referencedColumns == null)
		{
			throw new RuntimeException("Cannot pushdown a null selectList!");
		}
		
		columnsToProject = DBUtils.extractReferencedIdentifiers(referencedColumns, this.correlation);
	}

	@Override
	public void applySelectList(ArrayList<Expression> select)
	{
		this.evaluateSelectList = true;
		this.select = select;
	}

	@Override
	public void pushdownJoinPredicates(ArrayList<Expression> joinWhereClauses)
	{
		// no op
	}
}
